package org.gorson.executor;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.io.CsvReader;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.JoinedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;
import org.gorson.BO.AssetDetailBO;
import org.gorson.function.*;
import org.gorson.sink.SinkToMysqlPersonAsset;

import java.nio.file.Files;
import java.util.*;

public class FlinkSinkPersonAssetDetailToMysqlDemo {

    public static void main(String[] args) throws Exception {
        //使用本地模式并开启WebUI
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081-8082");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(1);
        String cmbFilePath = "C:\\Users\\Leon\\Desktop\\2021全年账单\\招商_2021.txt";
        SingleOutputStreamOperator<AssetDetailBO> source1 = env
                .readFile(new TextInputFormat(null), cmbFilePath, FileProcessingMode.PROCESS_ONCE, 10)
                .process(new ProcessCmbToAssetDetailFunction())
                .filter(new FilterCmbFunction());
//        source1.print("cmbSource =");



        String alipayFilePath = "C:\\Users\\Leon\\Desktop\\2021全年账单\\alipay_record_20220117_112758.csv";
//        String alipayFilePath = "C:\\Users\\Leon\\Desktop\\2021全年账单\\支付宝_2021.txt";
        SingleOutputStreamOperator<AssetDetailBO> source2 = env
                .readFile(new TextInputFormat(null), alipayFilePath, FileProcessingMode.PROCESS_ONCE, 10)
                .process(new ProcessAlipayToAssetDetailFunction())
                .filter(new FilterAlipayFunction());
//        source2.print("alipay = ");

        String weChatFilePath1 = "C:\\Users\\Leon\\Desktop\\2021全年账单\\2022微信账单\\微信支付账单(20210101-20210401).csv";
        String weChatFilePath2 = "C:\\Users\\Leon\\Desktop\\2021全年账单\\2022微信账单\\微信支付账单(20210402-20210630).csv";
        String weChatFilePath3 = "C:\\Users\\Leon\\Desktop\\2021全年账单\\2022微信账单\\微信支付账单(20210701-20210930).csv";
        String weChatFilePath4 = "C:\\Users\\Leon\\Desktop\\2021全年账单\\2022微信账单\\微信支付账单(20211001-20220101).csv";
        DataStreamSource<String> weChatSource1 = env
                .readFile(new TextInputFormat(null), weChatFilePath1, FileProcessingMode.PROCESS_ONCE, 10);
        DataStreamSource<String> weChatSource2 = env
                .readFile(new TextInputFormat(null), weChatFilePath2, FileProcessingMode.PROCESS_ONCE, 10);
        DataStreamSource<String> weChatSource3 = env
                .readFile(new TextInputFormat(null), weChatFilePath3, FileProcessingMode.PROCESS_ONCE, 10);
        DataStreamSource<String> weChatSource4 = env
                .readFile(new TextInputFormat(null), weChatFilePath4, FileProcessingMode.PROCESS_ONCE, 10);

        SingleOutputStreamOperator<AssetDetailBO> source3 = weChatSource1
                .union(weChatSource2)
                .union(weChatSource3)
                .union(weChatSource4)
                .process(new ProcessWechatToAssetDetailFunction())
                .filter(new FilterWechatFunction())
                .process(new ProcessFunction<AssetDetailBO, AssetDetailBO>() {
                    @Override
                    public void processElement(AssetDetailBO assetDetailBO, Context context, Collector<AssetDetailBO> collector) throws Exception {
                        assetDetailBO.setTransDate(assetDetailBO.getTransDate().substring(0,assetDetailBO.getTransDate().indexOf(" ")));
                        collector.collect(assetDetailBO);
                    }
                });

//        source3.print("wechat = ");
        DataStream<AssetDetailBO> union = source1
                .union(source2)
                .union(source3);
        union.print("union = ");

//        Properties properties = new Properties();
//        properties.setProperty("bootstrap.servers", "localhost:9092");
        union.addSink(new SinkToMysqlPersonAsset());
        env.execute("");
    }

}
